package com.atguigu.flink.sql.time;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.types.DataType;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/26 0026 16:52
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 使用TableAPI声明时间属性
 */
public class Flink01_TableApiTime {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                //       .column("user", DataTypes.STRING())
                .column("user", "STRING")
                .column("url", "STRING")
                .column("ts", "BIGINT")
                //处理时间列
                .columnByExpression("pt", "PROCTIME()")
                //事件时间,毫秒就写3，秒就写0
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts, 3)")
                //沿用流中的水位线
                //.watermark("et" , "source_watermark()")
                .watermark("et", "et - INTERVAL '0' SECOND")
                .build();


        //流转表
        Table table = tableEnv.fromDataStream(ds, schema);

        table.printSchema();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
